package com.learn.kotlinlearndemolist.coroutinesdemo

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.async
import kotlinx.coroutines.cancelChildren
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.reduce
import kotlinx.coroutines.launch
import kotlinx.coroutines.newSingleThreadContext
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import kotlinx.coroutines.withTimeoutOrNull
import java.util.concurrent.atomic.AtomicInteger
import kotlin.system.measureTimeMillis

/**
 * author: ts
 * created on: 2025/2/13 上午8:53
 * description:
 */

fun testPrev() = runBlocking {
    delay(1500L)
    println("testPrev Hello World!")
}

fun testFirst() = runBlocking {
    launch {
        delay(3000L)
        println("testFirst World!")
    }
    println("testFirst Hello")
}

fun testDivide() {
    println("--------------------------------------")
}

fun testDivideLine() {
    println("**************************************")
}

fun testSecond() = runBlocking {
    launch {
        doWorld("testSecond")
    }
    println("testSecond Hello")
}

fun testFourth() {
    GlobalScope.launch {
        launch {
            doHelloWorld("testFourth")
        }
        println("testFourth Hello")
    }
}

fun testFifth() = runBlocking {
    val time = measureTimeMillis {
        val one = async { doSomethingOne() }
        val two = async { doSomethingTwo() }
        println("testFifth The answer is ${one.await() + two.await()}")
    }
    println("testFifth Completed in $time ms")
}

fun testSixth() = runBlocking {
    val time = measureTimeMillis {
        val one = async(start = CoroutineStart.LAZY) { doSomethingOne() }
        val two = async(start = CoroutineStart.LAZY) { doSomethingTwo() }
        one.start()
        two.start()
        println("testSixth The answer is ${one.await() + two.await()}")
    }
    println("testSixth Completed in $time ms")
}

fun testSeventh() {
    val time = measureTimeMillis {
        val one = doSomethingOneAsync()
        val two = doSomethingTwoAsync()
        runBlocking {
            println("testSeventh The answer is ${one.await() + two.await()}")
        }
    }
    println("testSeventh Completed in $time ms")
}

suspend fun doSomethingOne(): Int {
    delay(1000L)
    return 12
}

suspend fun doSomethingTwo(): Int {
    delay(1000L)
    return 23
}

@OptIn(DelicateCoroutinesApi::class)
fun doSomethingOneAsync() = GlobalScope.async {
    doSomethingOne()
}

@OptIn(DelicateCoroutinesApi::class)
fun doSomethingTwoAsync() = GlobalScope.async {
    doSomethingTwo()
}

fun testEighth() = runBlocking {
    val time = measureTimeMillis {
        println("testEighth The answer is ${coroutineSum()}")
    }
    println("testEighth Completed in $time ms")
}

suspend fun coroutineSum(): Int = coroutineScope {
    val one = async { doSomethingOne() }
    val two = async { doSomethingTwo() }
    one.await() + two.await()
}

fun testNinth() = runBlocking {
    try {
        failedCoroutineSum()
    } catch (e: ArithmeticException) {
        println("testNinth Computation failed with ArithmeticException")
    }
}

fun simple(): Flow<Int> = flow {
    println("Flow started")
    for (i in 1..3) {
        delay(100L)
        emit(i)
    }
}

fun testTenth() = runBlocking {
    launch {
        for (k in 1..3) {
            println("I'm not blocked $k")
            delay(100)
        }
    }
    simple().collect { value -> println(value) }
}

fun testEleventh() = runBlocking {
    println("Calling simple function...")
    val flow = simple()
    println("Calling collect...")
    flow.collect { value -> println(value) }
    println("Calling collect again...")
    flow.collect { value -> println(value) }
}

fun testTwelve() = runBlocking {
    withTimeoutOrNull(250) {
        simpleEmit().collect { value -> println(value) }
    }
    println("Done")
}

fun testThirteen() = runBlocking {
    (1..3).asFlow().collect { value -> println("testThirteen $value") }
}

suspend fun performRequest(request: Int): String {
    delay(1000)
    return "request $request"
}

fun testFourteenth() = runBlocking {
    (1..3).asFlow().map { request -> performRequest(request) }
        .collect { response -> println("testFourteenth $response") }

    val sum = (1..5).asFlow().map { it * it }.reduce { accumulator, value ->
        println("accumulator $accumulator value $value")
        accumulator + value
    }
    println(sum)

    (1..5).asFlow().filter {
        println("Filter $it")
        it % 2 == 0
    }.map {
        println("Map $it")
        "string $it"
    }.collect {
        println("Collect $it")
    }

    simpleFlowOn().collect {
        log("Collected $it")
    }
}

fun simpleFlowOn(): Flow<Int> = flow {
    for (i in 1..3) {
        Thread.sleep(100)
        log("Emitting $i")
        emit(i)
    }
}.flowOn(Dispatchers.Default)

fun simpleEmit(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100L)
        println("Emitting $i")
        emit(i)
    }
}

suspend fun failedCoroutineSum(): Int = coroutineScope {
    val one = async {
        try {
            delay(Long.MAX_VALUE)
            42
        } finally {
            println("First child was cancelled")
        }
    }
    val two = async<Int> {
        println("Second child throw an exception")
        throw ArithmeticException()
    }
    one.await() + two.await()
}

fun testThird() = runBlocking {
    doHelloWorld("testThird")
}

suspend fun doWorld(msg: String) {
    delay(2000L)
    println("$msg World!")
}

suspend fun doHelloWorld(msg: String) = coroutineScope {
    launch {
        delay(1000L)
        println("$msg World!")
    }
    println("$msg Hello")
}

fun testRepeat() = runBlocking {
    repeat(50_000) {
        launch {
            delay(1000L)
            print(".")
        }
    }
}

fun testChannel() = runBlocking<Unit> {
    var channel = Channel<String>()

    launch {
        channel.send("launch A1")
        channel.send("launch A2")
        log("launch A done")
    }

    launch {
        channel.send("launch B1")
        log("launch B done")
    }

    launch {
        repeat(3) {
            var result = channel.receive()
            log(result)
        }
    }
}

fun log(msg: Any?) {
    println("[${Thread.currentThread().name}] $msg")
}

fun testPip() = runBlocking {
    var numbers = produceNumbers()
    var squares = square(numbers)
    repeat(5) {
        println(squares.receive())
    }
    println("testPip done!")
    coroutineContext.cancelChildren()
}

fun testPip2() = runBlocking {
    var cur = numbersFrom(2)
    repeat(10) {
        val prime = cur.receive()
        println(prime)
        cur = filter(cur, prime)
    }
    coroutineContext.cancelChildren()
}

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++)
}

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.square(number: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (x in number) send(x * x)
}

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.numbersFrom(start: Int) = produce {
    var x = start
    while (true) send(x++)
}

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.filter(number: ReceiveChannel<Int>, prime: Int) = produce {
    for (x in number) {
        if (x % prime != 0) send(x)
    }
}

suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 10
    val k = 100
    val time = measureTimeMillis {
        coroutineScope {
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    println("massiveRun ${n * k} actions in $time ms")
}

var counter = AtomicInteger()

val counterContext = newSingleThreadContext("CounterContext")

var contextCounter = 0

val mutex = Mutex()

var mutexCounter = 0
/**
 * 共享可变状态
 */
fun testShared() = runBlocking {
    withContext(Dispatchers.Default) {
        massiveRun {
            counter.incrementAndGet()
        }
    }
    println("testShared Counter = $counter")

    withContext(counterContext) {
        massiveRun {
            contextCounter++
        }
    }
    println("testShared contextCounter = $contextCounter")

    withContext(Dispatchers.Default) {
        massiveRun {
            mutex.withLock {
                mutexCounter++
            }
        }
    }
    println("testShared mutexCounter = $mutexCounter")
}